package com.shuhe.domain.dbTodb;

import com.shuhe.JettyServer;
import com.shuhe.common.MyDateUtils;
import com.shuhe.common.Tools;
import com.shuhe.domain.mxgc.Cqlb;
import com.shuhe.domain.mxgc.MxUserSjjrTaskResponseBody;
import com.shuhe.metastore.DBJrSql;
import com.shuhe.metastore.DBSql;
import com.shuhe.metastore.DbInflux;
import com.shuhe.metastore.MetaMysql;
import org.influxdb.InfluxDB;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.ResultSet;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.HashMap;

import static com.shuhe.common.CommonFunction.closeDb;
import static com.shuhe.common.CommonFunction.closeDbInflux;
import static com.shuhe.common.Tools.getConnect;


public class DbToDb {
    private static Logger LOG = LoggerFactory.getLogger(DbToDb.class);

    public static void execute(MxUserSjjrTaskResponseBody mxUserSjjrTaskResponseBody, String cronFun, String batchId
                               ) throws Exception {

        ArrayList<DbConnect> dbConnects = new ArrayList<DbConnect>();
        Connection metaConn = null;
        try {
            metaConn = MetaMysql.connect();
            //定时
            int threadNum = 0;
            String whereisTask = "";
            //自动任务，批量，走多线程；单个任务走单线程
            if (cronFun.equals("auto")) {
                threadNum = Integer.parseInt(JettyServer.globalProperties.getProperty("thread.num"));
                whereisTask = " and task_type='" + mxUserSjjrTaskResponseBody.getTaskType() + "'";
            }
            if (cronFun.equals("hand")) {
                whereisTask = " and  id='" + mxUserSjjrTaskResponseBody.getId() + "'";
            }
            String sql = "select count(*) from mxgc_user_task_sjjr where task_status='run'" + whereisTask;
            ResultSet rs = MetaMysql.querySql(metaConn, sql);
            rs.next();
            if (rs.getInt(1) == 0) {
                LOG.info("没有需要执行的任务, 此时可以关闭进程");
                Thread.sleep(100);
                return;
            }
            ArrayList<Thread> dbThreads = new ArrayList<Thread>();
            if (cronFun.equals("hand")) {
                sql = "select * from (\n" + "SELECT a.id,a.datasource_id,substr(a.table_name,instr(a.table_name,'.')" + "+1) table_name,"
                        + "a.table_name full_table_name,a.task_name,a.task_status,a.run_cron,a" + ".cron_type,a" + ".timeadd_column,\n"
                        + "a.timecolumn_format,a.start_time,a.new_time,a" + ".modify_time,a.desttable_name desttable_name,"
                        + "substr(a.desttable_name,1,instr" + "(desttable_name,'.')-1) dest_schema_name,"
                        + "substr(a.desttable_name,instr(a" + ".desttable_name,'.')+1) dest_table_name,a.task_type,a.user_name,\n"
                        + "a.timeadd_interval,a.no_task_time_begin,a.no_task_time_end,b.db_type,b.ip,b.source_port,b.db_name,\n"
                        + "b.db_user_name,b.PASSWORD,substr(a.table_name,1,instr(table_name,'.')-1) " + "schema_name \n"
                        + "FROM mxgc_user_task_sjjr a,mxgc_user_sjygl b WHERE a.datasource_id=b.id ) a where  "
                        + whereisTask.replace("and", "");
            } else {
                sql = "select * from (\n" + "SELECT a.id,a.datasource_id,substr(a.table_name,instr(a.table_name,'.')+1) table_name,"
                        + "a.table_name full_table_name,a.task_name,a.task_status,a.run_cron,"
                        + "a.cron_type,a.timeadd_column,a.timecolumn_format,a.start_time,a.new_time,"
                        + "a.modify_time,a.desttable_name desttable_name,substr(a.desttable_name,1,instr(desttable_name,'.')-1) dest_schema_name,"
                        + "substr(a.desttable_name,instr(" + "a.desttable_name,'.')+1) dest_table_name,a.task_type,a.user_name,"
                        + "a.timeadd_interval,a.no_task_time_begin,a.no_task_time_end,b.db_type,b.ip,b.source_port,"
                        + "b.db_name,b.db_user_name,b.PASSWORD,substr(a.table_name,1,instr(table_name,'.')-1) schema_name "
                        + " FROM mxgc_user_task_sjjr a,mxgc_user_sjygl b WHERE " + "a.datasource_id=b.id  ) "
                        + "a where a.task_status='run' order by a.task_type desc";
            }
            rs = MetaMysql.querySql(metaConn, sql);
            while (rs.next()) {
                Cqlb cqlb = new Cqlb();
                cqlb.setCtime(MyDateUtils.getNowDate());
                cqlb.setBatch_id(batchId);
                cqlb.setId(rs.getString("id"));
                cqlb.setDatasource_id(rs.getString("datasource_id"));
                cqlb.setTable_name(rs.getString("table_name"));
                cqlb.setFull_table_name(rs.getString("full_table_name"));
                cqlb.setTask_name(rs.getString("task_name"));
                cqlb.setTask_status(rs.getString("task_status"));
                cqlb.setRun_cron(rs.getString("run_cron"));
                cqlb.setCron_type(rs.getString("cron_type"));
                cqlb.setTimeadd_column(rs.getString("timeadd_column"));
                cqlb.setTimecolumn_Format(rs.getString("timecolumn_format"));
                cqlb.setStart_time(rs.getString("start_time"));
                cqlb.setNew_time(rs.getString("new_time"));   //ene_time
                cqlb.setModify_time(rs.getString("modify_time"));
                cqlb.setDesttable_name(rs.getString("desttable_name"));
                cqlb.setTask_type(rs.getString("task_type"));
                cqlb.setUser_name(rs.getString("user_name"));
                cqlb.setTimeadd_interval(rs.getString("timeadd_interval"));
                cqlb.setNo_task_time_begin(rs.getString("no_task_time_begin"));
                cqlb.setNo_task_time_end(rs.getString("no_task_time_end"));
                cqlb.setDb_type(rs.getString("db_type"));
                cqlb.setDb_ip(rs.getString("ip"));
                cqlb.setDb_port(rs.getString("source_port"));
                cqlb.setDb_name(rs.getString("db_name"));
                cqlb.setDb_user(rs.getString("db_user_name"));
                cqlb.setDb_password(rs.getString("password"));
                cqlb.setSchema_name(rs.getString("schema_name"));
                cqlb.setDest_schema_name(rs.getString("dest_schema_name"));
                cqlb.setDest_table_name(rs.getString("dest_table_name"));
                cqlb.setCronFun(cronFun);
                Calendar cal = Calendar.getInstance();
                if (!cqlb.getNo_task_time_begin().equals("") && !cqlb.getNo_task_time_end().equals("")) {
                    if (cal.get(Calendar.HOUR_OF_DAY) >= Integer.parseInt(cqlb.getNo_task_time_begin()) && cal.get(Calendar.HOUR_OF_DAY) <= Integer.parseInt(cqlb.getNo_task_time_end())) {
                        LOG.info("任务:" + cqlb.getId() + ",此时间不执行, current hour:" + cal.get(Calendar.HOUR_OF_DAY));
                        break;
                    }
                }
                InfluxDB sourInfluxDB = null;
                Connection sourConn = null;
                Connection destConn = null;
                DbConnect dbConnect = new DbConnect();
                HashMap<String, String> hashMaps = Tools.getConnect(cqlb.getDb_type(), cqlb.getDb_ip(),
                        cqlb.getDb_port(), cqlb.getDb_name());
                String url = hashMaps.get("url");
                boolean ifExist = false;
                for (int i = 0; i < dbConnects.size(); i++) {
                    if (url.equals(dbConnects.get(i).getUrl()) && cqlb.getTask_type().equals(dbConnects.get(i).getTaskType())) {
                        ifExist = true;
                        dbConnect = dbConnects.get(i);
                    }
                }
                if (!ifExist) {
                    String driver = hashMaps.get("className");
                    if (cqlb.getTask_type().equals("stg")) {
                        //接入
                        if (cqlb.getDb_type().toLowerCase().equals("influxdb")) {
                            sourInfluxDB = DbInflux.influxDbBuild(url, cqlb.getDb_user(), cqlb.getDb_password());
                        } else {
                            sourConn = DBSql.connectByAll(url, cqlb.getDb_user(), cqlb.getDb_password(), driver);
                            destConn = DBSql.connect();
                            if (JettyServer.globalProperties.getProperty("db.type").equals("gbase")) {
                                destConn = DBJrSql.connect();
                            }
                        }
                    } else {
                        //推送
                        destConn = DBSql.connectByAll(url, cqlb.getDb_user(), cqlb.getDb_password(), driver);
                        sourConn = DBSql.connect();
                        if (JettyServer.globalProperties.getProperty("db.type").equals("gbase")) {
                            sourConn = DBJrSql.connect();
                        }
                        cqlb.setDb_type(JettyServer.globalProperties.getProperty("db.type"));
                    }
                    dbConnect.setUrl(url);
                    dbConnect.setTaskType(cqlb.getTask_type());
                    dbConnect.setSourInfluxDB(sourInfluxDB);
                    dbConnect.setSourConn(sourConn);
                    dbConnect.setDestConn(destConn);
                    dbConnect.setMetaConn(metaConn);
                    dbConnects.add(dbConnect);
                }


                if (cqlb.getDb_type().toLowerCase().equals("influxdb")) {
                    dbThreads.add(new Thread(new DbToDbThread(cqlb, dbConnect)));
                } else {
                    dbThreads.add(new Thread(new DbToDbThread(cqlb, dbConnect)));
                }
                dbThreads.get(dbThreads.size() - 1).start();
                while (true) {
                    int run_num = 0;
                    for (int i = 0; i < dbThreads.size(); i++) {
                        if (!dbThreads.get(i).getState().equals(Thread.State.TERMINATED)) {
                            run_num = run_num + 1;
                        }
                    }
                    //并发数量控制
                    if (run_num > threadNum) {
                        LOG.info("正在运行的任务数量" + run_num);
                        Thread.sleep(100);
                    } else {
                        break;
                    }
                }
            }
            while (true) {
                int run_num = 0;
                for (int i = 0; i < dbThreads.size(); i++) {
                    if (!dbThreads.get(i).getState().equals(Thread.State.TERMINATED)) {
                        run_num = run_num + 1;
                    }
                }
                if (run_num > 0) {
                    LOG.info("正在运行的任务数量" + run_num);
                    Thread.sleep(100);
                } else {
                    break;
                }
            }
            LOG.info("完成一轮任务");
            sql = "insert into mxgc_user_task_log_sjjr_zt(nyr, zt,batch_id) values(CURDATE(), 'end','" + batchId + "')";
            MetaMysql.execSql(metaConn, sql);
            Thread.sleep(100);
        } finally {
            for (int i = 0; i < dbConnects.size(); i++) {
                closeDbInflux(dbConnects.get(i).getSourInfluxDB());
                closeDb(dbConnects.get(i).getSourConn());
                closeDb(dbConnects.get(i).getDestConn());
            }
            closeDb(metaConn);
        }
    }
}
